package f3source.s2customsource;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;
import java.sql.Date;
import java.util.concurrent.ThreadLocalRandom;

public class cussoufun implements SourceFunction<s131orders> {
    private Boolean running = true;

    @Override
    public void run(SourceContext<s131orders> ordersSC) throws Exception {
        Integer oid = 0;
        Random random = new Random();
        String[] status = {"paid", "pending", "shipped", "completed", "failed", "canceled", "returned"};


        long startMillis = new Date(2000 - 1990,0,1).getTime();
        long endMillis = new Date(2025 - 2000, 11, 31).getTime();

        while (running) {
            long randomMillis = ThreadLocalRandom.current().nextLong(startMillis, endMillis);
            ordersSC.collect(new s131orders(
                    oid++,
                    100 + random.nextInt(100),
                    300 + (long) random.nextInt(200),
                    status[random.nextInt(status.length)],
                    new Date(randomMillis)
            ));
            Thread.sleep(1000);
        }

    }
    @Override
    public void cancel () {
        running = false;
    }
}
